package com.allwees.core.mq.rabbitmq.config;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.allwees.core.mq.rabbitmq.MqException;
import com.allwees.core.mq.rabbitmq.MqUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;

/**
 * RabbitMq配置类，支持按配置创建Exchange，Queue和Binding
 *
 * @author yangylsky
 * @since 2019-11-18
 *
 */
@Slf4j
@Data
@ConfigurationProperties(prefix = "mq.rabbitmq")
public class RabbitMqConfig {
    private static RoutingInfo DEFAULT_ROUTING_INFO;

    @Autowired
    private RabbitProperties properties;

    /**
     * 批量处理消息Size
     */
	private int batchSize = 100;

	/**
	 * 批量处理消息超时时间，单位ms（如果队列未满但达到超时时间，也执行批处理）
	 */
	private long batchTimeout = 2000;

    private List<ExchangeInfo> exchangeInfo;

    private List<QueueInfo> queueInfo;

    private List<BindingInfo> bindingInfo;

	private List<RoutingInfo> routingInfo;

	@Autowired
	private AmqpAdmin amqpAdmin;

	private List<Exchange> exchanges = new ArrayList<>();

	private List<Queue> queues = new ArrayList<>();

	@PostConstruct
	private void _init() {
		try {
			_initExchangeInfos();
			_initQueueInfos();
			_initBindingInfos();
			log.info("RabbitMqConfig init successfullly");
		} catch(Exception e) {
			log.error("RabbitMqConfig init failed", e);
		}
	}

	private void _initExchangeInfos() {
		if(CollectionUtils.isNotEmpty(exchangeInfo)) {
			for(ExchangeInfo info : exchangeInfo) {
				_declareExchange(info);
			}
		}
	}

	private void _initQueueInfos() {
		if(CollectionUtils.isNotEmpty(queueInfo)) {
			for(QueueInfo info : queueInfo) {
				_declareQueue(info);
			}
		}
	}

	private void _initBindingInfos() {
		if(CollectionUtils.isNotEmpty(bindingInfo)) {
			for(BindingInfo info : bindingInfo) {
				_bind(info);
			}
		}
	}

	private void _declareExchange(ExchangeInfo info) {
		ExchangeBuilder exBuilder = new ExchangeBuilder(info.getName(), info.getType()).durable(info.isDurable());
		if(info.isAutoDelete()) {
			exBuilder.autoDelete();
		}
		Exchange exchange = exBuilder.build();
		exchanges.add(exchange);
		amqpAdmin.declareExchange(exchange);
		log.info("Exchange[{}: {}] declared", exchange.getType(), exchange.getName());
	}

	private void _declareQueue(QueueInfo info) {
		QueueBuilder queueBuilder = info.isDurable() ? QueueBuilder.durable(info.getName()) : QueueBuilder.nonDurable(info.getName());
		if(info.isAutoDelete()) {
			queueBuilder.autoDelete();
		}
		if(info.isExclusive()) {
			queueBuilder.exclusive();
		}
		Queue queue = queueBuilder.withArguments(info.getArguments()).build();
		queues.add(queue);
		amqpAdmin.declareQueue(queue);
		log.info("===> Queue[{}] declared", info.getName());
	}


	private void _bind(BindingInfo info) {
		Binding binding = new Binding(info.getDestination(), info.getDestinationType(), info.getExchange(), info.getRoutingKey(), info.getArguments());
		amqpAdmin.declareBinding(binding);
		log.info("Binding[{}({}) -> {}({}), args:{}] declared", info.getExchange(), info.getRoutingKey(), info.getDestination(), info.getDestinationType(),
				info.getArguments());
	}



	public boolean queueExist(String name) {
		return amqpAdmin.getQueueProperties(name) != null;
	}

	public void checkQueue(String name) {
		if(!queueExist(name)) {
			throw new MqException("Queue[" + name + "] is not exist");
		}
	}

	public RoutingInfo findRoutingInfo(String bizType) {
		for(RoutingInfo routingInfo : routingInfo) {
			if(routingInfo.getBizType().equals(bizType)) {
				return routingInfo;
			}
		}
		log.warn("cannot find RoutingInfo[bizType:{}], use DEFAULT_ROUTING_INFO", bizType);
		return DEFAULT_ROUTING_INFO;
	}

	private ObjectMapper jsonObjectMapper() {
		return new Jackson2ObjectMapperBuilder()
				.indentOutput(true)
				.serializationInclusion(JsonInclude.Include.NON_NULL)
				.build();
	}

	@Bean
	public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(CachingConnectionFactory connectionFactory, ThreadPoolTaskExecutor taskExecutor,
			RetryTemplate retryTemplate) {
		connectionFactory.setExecutor(taskExecutor);
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factory.setConnectionFactory(connectionFactory);
		factory.setMessageConverter(new Jackson2JsonMessageConverter(jsonObjectMapper()));
		factory.setErrorHandler(new RabbitListenerErrorHandler());
		factory.setAdviceChain(
				RetryInterceptorBuilder
					.stateless()
						.recoverer(new RejectAndDontRequeueRecoverer())
						.retryOperations(retryTemplate)
					.build()
			);
		return factory;
	}

	@Bean
	private RetryTemplate retryTemplate(ExponentialBackOffPolicy backOffPolicy, SimpleRetryPolicy retryPolicy) {
		RetryTemplate retryTemplate = new RetryTemplate();
		retryTemplate.setBackOffPolicy(backOffPolicy);
		retryTemplate.setRetryPolicy(retryPolicy);
		return retryTemplate;
	}

	@Bean
	private ExponentialBackOffPolicy backOffPolicy() {
		ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
		backOffPolicy.setInitialInterval(properties.getListener().getSimple().getRetry().getInitialInterval().toMillis());
		backOffPolicy.setMaxInterval(properties.getListener().getSimple().getRetry().getMaxInterval().toMillis());
		backOffPolicy.setMultiplier(properties.getListener().getSimple().getRetry().getMultiplier());
		return backOffPolicy;
	}

	@Bean
	private SimpleRetryPolicy retryPolicy() {
		return new SimpleRetryPolicy(properties.getListener().getSimple().getRetry().getMaxAttempts());
	}

	@Bean
	public MqUtil mqUtil(RabbitTemplate rabbitTemplate) {
		DEFAULT_ROUTING_INFO = new RoutingInfo(rabbitTemplate.getExchange(), rabbitTemplate.getRoutingKey());
		return new MqUtil(this, rabbitTemplate);
	}
}
